package com.wyl.learn05;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.StopWatch;
import cn.hutool.core.util.StrUtil;
import com.rabbitmq.client.*;
import com.wyl.config.RabbitMqConfig;
import com.wyl.learn04.MessageBean;
import com.wyl.util.QueueArguments;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * ttl超时，进入死信队列 实战
 */
public class Producer05_1 {
    public static final String EXCHANGE_NAME = "exchange5";
    public static final String QUEUE_NAME = "queue5";

    public static final String EXCHANGE_NAME_DEAD_MSG = "exchange5_dead_msg";
    public static final String QUEUE_NAME_DEAD_MSG1 = "dead_msg_queue1";
    public static final String QUEUE_NAME_DEAD_MSG2 = "dead_msg_queue2";
    public static final String QUEUE_NAME_DEAD_MSG3 = "dead_msg_queue3";

    public static void main(String[] argv) throws Exception {
        /// 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMqConfig.IP);
        factory.setUsername(RabbitMqConfig.USERNAME);
        factory.setPassword(RabbitMqConfig.PASSWORD);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        /// 声明死信交换机
        // 如果死信交换机为BuiltinExchangeType.FANOUT，那么也不会理会路由，会直接广播
        channel.exchangeDeclare(EXCHANGE_NAME_DEAD_MSG, BuiltinExchangeType.TOPIC);

        /// 声明接收死信的队列，并绑定死信交换机
        channel.queueDeclare(QUEUE_NAME_DEAD_MSG1, false, false, false, null);
        channel.queueDeclare(QUEUE_NAME_DEAD_MSG2, false, false, false, null);
        channel.queueDeclare(QUEUE_NAME_DEAD_MSG3, false, false, false, null);
        // 死信交换机会把消息转换到跟他绑定了的队列，并绑定不同的路由，看看路由的效果
        // 预期效果：QUEUE_NAME_DEAD_MSG1 11条消息，QUEUE_NAME_DEAD_MSG2 0条消息，QUEUE_NAME_DEAD_MSG3有10条消息
        // 因为EXCHANGE_NAME_DEAD_MSG是扇出交换机，会广播给所有死信队列
        channel.queueBind(QUEUE_NAME_DEAD_MSG1, EXCHANGE_NAME_DEAD_MSG, "deadMsg1");
        channel.queueBind(QUEUE_NAME_DEAD_MSG2, EXCHANGE_NAME_DEAD_MSG, "deadMsg2");
        // 死信队列绑定死信交换机，那么routerKey是必填的，同时也不能是空字符串
        channel.queueBind(QUEUE_NAME_DEAD_MSG3, EXCHANGE_NAME_DEAD_MSG, "*");

        /// 声明普通交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

        /// 声明队列，并绑定普通交换机和死信交换机
        Map<String, Object> queueArgumentMap = new HashMap<>();
        queueArgumentMap.put(QueueArguments.XDeadLetterExchange.getArgumentName(), EXCHANGE_NAME_DEAD_MSG);
        // rabbitMq默认的死信交换机路由只能写死一个，如果想根据消息的routerKey来路由，需要安装插件
        queueArgumentMap.put(QueueArguments.XDeadLetterRoutingKey.getArgumentName(), "deadMsg1"); // 这个routingKey是让死信交换机路由的
        channel.queueDeclare(QUEUE_NAME, false, false, false, queueArgumentMap);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*"); // 只有topic交换机才能解析*？否则如果是定义为direct交换机的路由为*，消息的路由要真正为*才会被接收

        // 开启发布确认
        channel.confirmSelect();
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.是否批量确认。true 可以确认小于等于当前序列号的消息；false 确认当前序列号消息。在RabbitMQ中，Confirm模式的批量确认是由服务器自动进行的，无法通过配置或参数设置来控制。
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                // 返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                System.out.println("成功接收消息：");
                System.out.println(CollUtil.join(confirmed.values(), "\n"));
                // 清除该部分未确认消息
                confirmed.clear();

            } else {
                // 只清除当前序列号的消息
                String message = outstandingConfirms.remove(deliveryTag);
                System.out.println("成功接收消息：" + message);
            }
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布的消息" + message + "未被确认，序列号" + deliveryTag);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        List<MessageBean> messageBeans = new ArrayList<>();
        for (int i = 0; i < 11; i++) {
            MessageBean messageBean = new MessageBean("消息" + i + ": 10秒没被消费则进入死信")
                    .setId(i + "")
//                    .setQueueName(QUEUE_NAME) // 如果直接发给目标队列，那死信队列与死信交换机的routingKey也失效了
                    .setExchangeName(EXCHANGE_NAME)
                    .setRouterKey("deadMsg" + i) // 这个routingKey只是让普通交换机路由的
                    .setTtl(10 * 1000L);
            messageBeans.add(messageBean);
        }

        for (MessageBean messageBean : messageBeans) {
            Thread.sleep(2000);
            String message = messageBean.getMessage();
            // 消息确认
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            // 构建消息属性
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            if (messageBean.getTtl() != null) {
                builder.expiration(messageBean.getTtl()); // 消息的 TTL 为 10 秒
            }
            AMQP.BasicProperties properties = builder.build();

            // 发送到队列
            if (StrUtil.isNotBlank(messageBean.getQueueName())) {
                channel.basicPublish("", messageBean.getQueueName(), properties, message.getBytes(StandardCharsets.UTF_8));
            }
            // 发送到交换机
            else if (StrUtil.isAllNotBlank(messageBean.getExchangeName(), messageBean.getRouterKey())){
                channel.basicPublish(messageBean.getExchangeName(), messageBean.getRouterKey(), properties, message.getBytes(StandardCharsets.UTF_8));
            }
        }
    }
}